import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

/**
 * 数据持久化工具
 * 将Spark处理的结果保存到SQLite数据库
 */
object DataPersistence {
  
  // SQLite数据库连接配置
  private val jdbcUrl = "jdbc:sqlite:orders_system.db"
  private val driverClass = "org.sqlite.JDBC"
  
  /**
   * 获取数据库连接
   */
  def getConnection: Connection = {
    Class.forName(driverClass)
    DriverManager.getConnection(jdbcUrl)
  }
  
  /**
   * 保存订单原始数据
   */
  def saveOrderRawData(category: String, productName: String, quantity: Int, 
                       orderDate: String, customerRating: Int, isValid: String): Unit = {
    val sql = "INSERT INTO order_raw (category, product_name, quantity, order_date, customer_rating, is_valid) VALUES (?, ?, ?, ?, ?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, category)
      stmt.setString(2, productName)
      stmt.setInt(3, quantity)
      stmt.setString(4, orderDate)
      stmt.setInt(5, customerRating)
      stmt.setString(6, isValid)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 更新有效/无效订单统计
   */
  def updateValidInvalidStats(isValid: String, count: Long): Unit = {
    val sql = "INSERT OR REPLACE INTO order_valid_stats (is_valid, count_total) VALUES (?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, isValid)
      stmt.setLong(2, count)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 更新产品订单统计
   */
  def updateProductOrderStats(productName: String, isValid: String, count: Long): Unit = {
    val sql = "INSERT OR REPLACE INTO product_order_stats (product_name, is_valid, count_total) VALUES (?, ?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, productName)
      stmt.setString(2, isValid)
      stmt.setLong(3, count)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 更新类别订单统计
   */
  def updateCategoryOrderStats(category: String, count: Long): Unit = {
    val sql = "INSERT OR REPLACE INTO category_order_stats (category, count_total) VALUES (?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, category)
      stmt.setLong(2, count)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 更新类别有效/无效统计
   */
  def updateCategoryValidStats(category: String, isValid: String, count: Long): Unit = {
    val sql = "INSERT OR REPLACE INTO category_valid_stats (category, is_valid, count_total) VALUES (?, ?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, category)
      stmt.setString(2, isValid)
      stmt.setLong(3, count)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 更新类别-产品统计
   */
  def updateCategoryProductStats(category: String, productName: String, isValid: String, count: Long): Unit = {
    val sql = "INSERT OR REPLACE INTO category_product_stats (category, product_name, is_valid, count_total) VALUES (?, ?, ?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, category)
      stmt.setString(2, productName)
      stmt.setString(3, isValid)
      stmt.setLong(4, count)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 批量保存数据到数据库
   */
  def batchSaveToDatabase(data: Array[(String, Long)], updateFunc: (String, Long) => Unit): Unit = {
    data.foreach { case (key, count) =>
      try {
        updateFunc(key, count)
      } catch {
        case e: Exception =>
          println(s"保存数据失败: $key -> $count, 错误: ${e.getMessage}")
      }
    }
  }
  
  /**
   * 将用户评分数据保存到数据库
   */
  def saveUserItemRating(userId: String, itemId: String, rating: Double): Unit = {
    val sql = "INSERT OR REPLACE INTO user_item_ratings (user_id, item_id, rating) VALUES (?, ?, ?)"
    
    val conn = getConnection
    try {
      val stmt = conn.prepareStatement(sql)
      stmt.setString(1, userId)
      stmt.setString(2, itemId)
      stmt.setDouble(3, rating)
      stmt.executeUpdate()
      stmt.close()
    } finally {
      conn.close()
    }
  }
  
  /**
   * 测试数据库连接
   */
  def testConnection(): Boolean = {
    try {
      val conn = getConnection
      val isValid = conn.isValid(5)
      conn.close()
      isValid
    } catch {
      case e: Exception =>
        println(s"数据库连接测试失败: ${e.getMessage}")
        false
    }
  }
} 